package com.zjh.es;


import java.util.concurrent.Flow;

public class SimpleReactiveDemo {

    public static void main(String[] args) {

        // 1️⃣ 定义 Publisher（发布者）
        Flow.Publisher<Integer> publisher = new Flow.Publisher<>() {

            @Override
            public void subscribe(Flow.Subscriber<? super Integer> subscriber) {
                // 每次订阅时创建一个 Subscription（订阅关系）
                Flow.Subscription subscription = new Flow.Subscription() {
                    int count = 0;
                    boolean canceled = false;

                    @Override
                    public void request(long n) {
                        // 模拟数据发布
                        for (int i = 0; i < n && !canceled; i++) {
                            if (count < 5) {
                                subscriber.onNext(count++); // 发布数据
                            } else {
                                subscriber.onComplete(); // 结束信号
                                break;
                            }
                        }
                    }

                    @Override
                    public void cancel() {
                        canceled = true;
                        System.out.println("订阅已取消");
                    }
                };

                // 通知订阅者：订阅建立成功
                subscriber.onSubscribe(subscription);
            }
        };

        // 2️⃣ 定义 Subscriber（订阅者）
        Flow.Subscriber<Integer> subscriber = new Flow.Subscriber<>() {

            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                System.out.println("✅ 订阅成功，开始请求数据...");
                subscription.request(1); // 首次请求1条数据
            }

            @Override
            public void onNext(Integer item) {
                System.out.println("📦 收到数据：" + item);
                try {
                    Thread.sleep(500); // 模拟处理延迟
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                subscription.request(1); // 处理完一条再请求下一条（背压）
            }

            @Override
            public void onError(Throwable throwable) {
                System.err.println("❌ 出错：" + throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("🎉 数据接收完毕！");
            }
        };

        // 3️⃣ 建立订阅关系
        publisher.subscribe(subscriber);
    }
}
